package com.jf.concurrent.queue.disruptor.demo;

import com.jf.concurrent.queue.disruptor.consumer.MessageEventHandler;
import com.jf.concurrent.queue.disruptor.model.MessageModel;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.util.concurrent.DefaultThreadFactory;

/**
 * 构造MQManager
 */
public class MQManager {

    public static RingBuffer<MessageModel> buidldRingBuffer() {
	// 定义用于创建事件处理的线程的线程工厂， Disruptor通过java.util.concurrent.ExecutorSerivce提供的线程来触发consumer的事件处理
	final DefaultThreadFactory threadFactory = new DefaultThreadFactory("consumer-pool");

	//指定事件工厂
	MessageEventFactory messageEventFactory = new MessageEventFactory();

	//指定RingBuffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
	int bufferSize = 1024 * 256;

	//单线程模式，获取额外的性能
	Disruptor<MessageModel> disruptor = new Disruptor<>(messageEventFactory, bufferSize, threadFactory,
		ProducerType.SINGLE, new BlockingWaitStrategy());

	//设置事件业务处理器---消费者
	disruptor.handleEventsWith(new MessageEventHandler());
	// 启动disruptor线程
	disruptor.start();

	//获取RingBuffer环，用于接取生产者生产的事件
	RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();

	return ringBuffer;
    }
}
